package org.apache.dubbo.common.io;

import java.io.IOException;
import java.io.InputStream;

/**
 * Stream utils.
 */
public class StreamUtils {

    private StreamUtils() {

    }

    public static InputStream limitedInputStream(final InputStream is,final int limit) throws IOException {
        return new InputStream() {

            private int position = 0, mark = 0;

            private final int mLimit = Math.min(limit,is.available());

            @Override
            public int read() throws IOException {
                if (position < limit) {
                    position ++;
                    return is.read();
                }
                return -1;
            }

            @Override
            public int read(byte[] b, int off, int len) throws IOException {
                if (b == null) {
                    throw new NullPointerException();
                }

                if (off < 0 || len < 0 || len > b.length - off) {
                    throw new IndexOutOfBoundsException();
                }

                if (position >= mLimit) {
                    return -1;
                }

                if (position + len > limit) {
                    len = mLimit - position;
                }

                if (len <= 0) {
                    return 0;
                }

                is.read(b,off,len);
                position += len;
                return len;
            }

            @Override
            public long skip(long len) throws IOException {
                if (position + len > mLimit) {
                    len = mLimit - position;
                }

                if (len <= 0) {
                    return 0;
                }

                is.skip(len);
                position += len;
                return len;
            }

            @Override
            public int available() throws IOException {
                return mLimit - position;
            }

            @Override
            public boolean markSupported() {
                return is.markSupported();
            }

            @Override
            public synchronized void mark(int readlimit) {
                is.mark(readlimit);
                mark = position;
            }

            @Override
            public synchronized void reset() throws IOException {
                is.reset();
                position = mark;
            }

            @Override
            public void close() throws IOException {
                is.close();
            }
        };
    }

    public static InputStream markSupportedInputStream(final InputStream is,final int markBufferSize) {
        if (is.markSupported()) {
            return is;
        }
        return new InputStream() {

            byte[] mMarkBuffer;

            boolean mInMarked = false;
            boolean mInReset = false;
            boolean mDry = false;
            private int mPosition = 0;
            private int mCount = 0;

            @Override
            public int read() throws IOException {
                if (!mInMarked) {
                    return is.read();
                }
                if (mPosition < mCount) {
                    byte b = mMarkBuffer[mPosition ++];
                    return b & 0xFF;
                }

                if (!mInReset) {
                    if (mDry) {
                        return -1;
                    }

                    if (mMarkBuffer == null) {
                        mMarkBuffer = new byte[markBufferSize];
                    }
                    if (mPosition >= markBufferSize) {
                        throw new IOException("Mark buffer is full!");
                    }

                    int read = is.read();

                    if (read == -1) {
                        mDry = true;
                        return -1;
                    }

                    mMarkBuffer[mPosition ++] = (byte) read;
                    mCount ++;

                    return read;
                }

                mInMarked = false;
                mInReset = false;
                mPosition = 0;
                mCount = 0;
                return is.read();
            }

            @Override
            public synchronized void mark(int readlimit) {
                mInMarked = true;
                mInReset = false;

                int count = mCount - mPosition;

                if (count > 0) {
                    System.arraycopy(mMarkBuffer, mPosition, mMarkBuffer, 0, count);
                    mCount = count;
                    mPosition = 0;
                }
            }

            @Override
            public synchronized void reset() throws IOException {
                if (!mInMarked) {
                    throw new IOException("should mark before reset!");
                }

                mInReset = true;
                mPosition = 0;
            }

            @Override
            public boolean markSupported() {
                return true;
            }

            @Override
            public int available() throws IOException {
                int available = is.available();

                if (mInMarked && mInReset) {
                    available += mCount - mPosition;
                }
                return available;
            }

            @Override
            public void close() throws IOException {
                is.close();
            }
        };
    }

    public static InputStream markSupportedInputStream(final InputStream in) {
        return markSupportedInputStream(in,1024);
    }

    public static void skipUnusedStream(InputStream is) throws IOException {
        if (is.available() > 0) {
            is.skip(is.available());
        }
    }
}
